package com.atguigu.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.TradeProvinceOrderBean;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.function.BeanToJsonMapFunction;
import com.atguigu.gmall.realtime.common.function.DimAsyncRichFunction;
import com.atguigu.gmall.realtime.common.util.DateFormatUtil;
import com.atguigu.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.operators.sort.DefaultInMemorySorterFactory;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * 从Kafka读取订单明细数据，过滤null数据并按照唯一键对数据去重，统计各省份各窗口订单数和订单金额，将数据写入Doris交易域省份粒度下单各窗口汇总表。
 */
public class DwsTradeProvinceOrderWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTradeProvinceOrderWindow().start(10020, 4, "dws_trade_province_order_window", Constant.TOPIC_DWD_TRADE_ORDER_DETAIL);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 从Kafka下单明细主题读取数据
        // 过滤null数据并转换数据结构  转换数据结构 JSONObject转换为实体类TradeProvinceOrderWindow。
        SingleOutputStreamOperator<TradeProvinceOrderBean> mapDs = stream.map(
                new RichMapFunction<String, TradeProvinceOrderBean>() {
                    @Override
                    public TradeProvinceOrderBean map(String value) throws Exception {
                        JSONObject jsonObj = JSON.parseObject(value);
                        HashSet<String> set = new HashSet<>();
                        set.add(jsonObj.getString("order_id"));
                        return TradeProvinceOrderBean.builder()
                                .orderDetailId(jsonObj.getString("id"))
                                .orderAmount(jsonObj.getBigDecimal("split_total_amount"))
                                .ts(jsonObj.getLong("ts") * 1000)
                                .orderIdSet(set)
                                .provinceId(jsonObj.getString("province_id"))
                                .build();
                    }
                }
        );
        // mapDs.print("MAP");

        /*// 按照唯一键去重
        SingleOutputStreamOperator<TradeProvinceOrderBean> distinctDs = mapDs.keyBy(TradeProvinceOrderBean::getOrderDetailId)
                .process(
                        new KeyedProcessFunction<String, TradeProvinceOrderBean, TradeProvinceOrderBean>() {
                            private ValueState<TradeProvinceOrderBean> lastBeanState;

                            @Override
                            public void open(Configuration parameters) throws Exception {
                                ValueStateDescriptor<TradeProvinceOrderBean> lastBeanStateDesc
                                        = new ValueStateDescriptor<>("lastBeanStateDesc", Types.POJO(TradeProvinceOrderBean.class));
                                //TTL
                                lastBeanStateDesc.enableTimeToLive(
                                        StateTtlConfig.newBuilder(Time.seconds(10)).build()
                                );
                                lastBeanState = getRuntimeContext().getState(lastBeanStateDesc);
                            }

                            @Override
                            public void processElement(TradeProvinceOrderBean bean, KeyedProcessFunction<String, TradeProvinceOrderBean, TradeProvinceOrderBean>.Context context, Collector<TradeProvinceOrderBean> collector) throws Exception {
                                // 状态+抵消
                                TradeProvinceOrderBean lastValue = lastBeanState.value();
                                if (lastValue == null) {
                                    lastBeanState.update(bean);
                                    collector.collect(bean);
                                } else {
                                    lastValue.setOrderAmount(bean.getOrderAmount().subtract(lastValue.getOrderAmount()));
                                    collector.collect(lastValue);
                                    lastBeanState.update(bean);
                                }
                            }
                        }
                );
        distinctDs.print("DISTINCT");*/

        SingleOutputStreamOperator<TradeProvinceOrderBean> distinctDs = mapDs.keyBy(TradeProvinceOrderBean::getOrderDetailId)
                .process(
                        new KeyedProcessFunction<String, TradeProvinceOrderBean, TradeProvinceOrderBean>() {
                            private ValueState<Boolean> lastBeanState;

                            @Override
                            public void open(Configuration parameters) throws Exception {
                                RuntimeContext runtimeContext = getRuntimeContext();
                                ValueStateDescriptor<Boolean> booleanValueStateDescriptor = new ValueStateDescriptor<Boolean>("lastBeanStateDesc", Types.BOOLEAN);
                                booleanValueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(20)).build());
                                lastBeanState = runtimeContext.getState(booleanValueStateDescriptor);
                            }

                            @Override
                            public void processElement(TradeProvinceOrderBean bean, KeyedProcessFunction<String, TradeProvinceOrderBean, TradeProvinceOrderBean>.Context context, Collector<TradeProvinceOrderBean> collector) throws Exception {
                                if (lastBeanState.value() == null) {
                                    lastBeanState.update(true);
                                    collector.collect(bean);
                                }
                            }
                        }
                );
        // distinctDs.print("DISTINCT");



        // 设置水位线
        SingleOutputStreamOperator<TradeProvinceOrderBean> watermarksDs = distinctDs.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TradeProvinceOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3L))
                        .withTimestampAssigner((bean, ts) -> bean.getTs())
                        .withIdleness(Duration.ofSeconds(120L))
        );


        // 按照省份ID分组 provinceId可以唯一标识数据。开窗聚合计算
        // 度量字段求和，并在窗口闭合后补充窗口起始时间、结束时间以及当前统计日期。
        SingleOutputStreamOperator<TradeProvinceOrderBean> reduceDs = watermarksDs.keyBy(TradeProvinceOrderBean::getProvinceId)
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(5L)))
                .reduce(
                        new ReduceFunction<TradeProvinceOrderBean>() {
                            @Override
                            public TradeProvinceOrderBean reduce(TradeProvinceOrderBean t1, TradeProvinceOrderBean t2) throws Exception {
                                t1.setOrderAmount(t1.getOrderAmount().add(t2.getOrderAmount()));
                                t1.getOrderIdSet().addAll(t2.getOrderIdSet());
                                return t1;
                            }
                        },
                        new ProcessWindowFunction<TradeProvinceOrderBean, TradeProvinceOrderBean, String, TimeWindow>() {
                            @Override
                            public void process(String key, ProcessWindowFunction<TradeProvinceOrderBean, TradeProvinceOrderBean, String, TimeWindow>.Context context, Iterable<TradeProvinceOrderBean> iterable, Collector<TradeProvinceOrderBean> collector) throws Exception {
                                TradeProvinceOrderBean bean = iterable.iterator().next();
                                String stt = DateFormatUtil.tsToDateTime(context.window().getStart());
                                String edt = DateFormatUtil.tsToDateTime(context.window().getEnd());
                                String curDate = DateFormatUtil.tsToDate(context.window().getStart());
                                bean.setStt(stt);
                                bean.setEdt(edt);
                                bean.setCurDate(curDate);
                                bean.setOrderCount((long) bean.getOrderIdSet().size());
                                collector.collect(bean);
                            }
                        }
                );
        // reduceDs.print("REDUCE");


        // 关联省份信息 补全省份名称字段。
        SingleOutputStreamOperator<TradeProvinceOrderBean> dimAsyncDs = AsyncDataStream.unorderedWait(
                reduceDs,
                new DimAsyncRichFunction<TradeProvinceOrderBean>() {
                    @Override
                    public String getRowKey(TradeProvinceOrderBean bean) {
                        return bean.getProvinceId();
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_province";
                    }

                    @Override
                    public void addDims(TradeProvinceOrderBean bean, JSONObject dimFromHbase) {
                        bean.setProvinceName(dimFromHbase.getString("name"));
                    }
                },
                20,
                TimeUnit.SECONDS
        );
        dimAsyncDs.print("ASYNC");


        // 写出到Doris。
        dimAsyncDs.map(
                new BeanToJsonMapFunction<>()
        ).sinkTo(FlinkSinkUtil.getDorisSink(Constant.DORIS_DATABASE, Constant.DWS_TRADE_PROVINCE_ORDER_WINDOW));

    }
}
